package com.partsoft.dits.integrate;



import java.io.UnsupportedEncodingException;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;

import com.partsoft.dits.DataBroadcaster;
import com.partsoft.dits.DataEdgeDescriptor;
import com.partsoft.dits.DataIntegrateBatch;
import com.partsoft.dits.DataIntegrateProcessor;
import com.partsoft.dits.DataMessageReceiver;
import com.partsoft.dits.DataReceiver;
import com.partsoft.dits.DataTransformer;
import com.partsoft.dits.log.Log;
import com.partsoft.dits.message.MessageEntity;
import com.partsoft.dits.message.MessageException;
import com.partsoft.dits.message.MessageProcessor;

/**
 * TODO 需要重新考虑
 */
@SuppressWarnings("unchecked")
public class BaseIntegrateReactor implements IntegrateReactor {

	private Map<Class<? extends DataEdgeDescriptor>, DataTransformer> dataTransformers = Collections.synchronizedMap(new HashMap<Class<? extends DataEdgeDescriptor>, DataTransformer>(5));
	
	private Set<DataBroadcaster> dataBroadcasters = Collections.synchronizedSet(new HashSet<DataBroadcaster>(5));
	
	private Set<DataReceiver> dataReceivers = Collections.synchronizedSet(new HashSet<DataReceiver>(5));
	
	private MessageProcessor messageProcessor = MessageProcessor.INSTANCE;
	
	public boolean supportIntegrateProcessor(Class<? extends DataIntegrateProcessor> processorClazz) {
		Assert.notNull(processorClazz);
		return ClassUtils.isAssignable(DataTransformer.class, processorClazz) ||
				ClassUtils.isAssignable(DataBroadcaster.class, processorClazz) ||
				ClassUtils.isAssignable(DataReceiver.class, processorClazz);
	}
	
	public boolean supportIntegrateProcessor(DataIntegrateProcessor processor) {
		Assert.notNull(processor);
		return supportIntegrateProcessor((Class<? extends DataIntegrateProcessor>) ClassUtils.getUserClass(processor));
	}
	
	public void addDataIntegrateProcessor(DataIntegrateProcessor processor) {
		Assert.notNull(processor);
		Class<? extends DataIntegrateProcessor> processorClazz = (Class<? extends DataIntegrateProcessor>) ClassUtils.getUserClass(processor);
		Assert.isTrue(supportIntegrateProcessor(processorClazz), String.format("不支持的数据处理器类型(%s)", processorClazz.getName()));
		if (ClassUtils.isAssignable(DataTransformer.class, processorClazz)) {
			DataTransformer transformer = (DataTransformer) processor;
			Class<? extends DataEdgeDescriptor> dataClazz = transformer.getSupportClass();
			if (dataTransformers.containsKey(dataClazz)) {
				Log.warn(String.format("数据描述类(%s)转换处理器已被注册，不能重复注册。", dataClazz.getName(), processor.toString()));
				return;
			}
			Log.warn(String.format("找到数据描述类(%s)转换处理器(%s)并成功注册。", dataClazz.getName(), processorClazz.getName()));
			dataTransformers.put(dataClazz, transformer);
		} else if (ClassUtils.isAssignable(DataBroadcaster.class, processorClazz)) {
			Log.warn(String.format("找到数据广播器(%s)并成功注册。", processorClazz.getName()));
			dataBroadcasters.add((DataBroadcaster) processor);
		} else if (ClassUtils.isAssignable(DataReceiver.class, processorClazz)) {
			Log.warn(String.format("找到数据接收器(%s)并成功注册。", processorClazz.getName()));
			dataReceivers.add((DataReceiver) processor);
		}
	}
	
	public void removeDataIntegrateProcessor(DataIntegrateProcessor processor) {
		Assert.notNull(processor);
		Class<? extends DataIntegrateProcessor> processorClazz = (Class<? extends DataIntegrateProcessor>) ClassUtils.getUserClass(processor);
		if (ClassUtils.isAssignable(DataTransformer.class, processorClazz)) {
			DataTransformer transformer = (DataTransformer) processor;
			Class<? extends DataEdgeDescriptor> dataClazz = transformer.getSupportClass();
			if (dataTransformers.containsKey(dataClazz)) {
				if (dataTransformers.get(dataClazz) == processor) {
					dataTransformers.remove(dataClazz);
				}
			}
		} else if (ClassUtils.isAssignable(DataBroadcaster.class, processorClazz)) {
			dataBroadcasters.remove(processor);
		} else if (ClassUtils.isAssignable(DataMessageReceiver.class, processorClazz)) {
			dataReceivers.remove(processor);
		}
	}
	
	protected void clearDataIntegrateProcessors() {
		for (DataIntegrateProcessor processor : getDataIntegrateProcessors()) {
			removeDataIntegrateProcessor(processor);
		}
	}
	
	public void setMessageProcessor(MessageProcessor messageProcessor) {
		this.messageProcessor = messageProcessor;
	}
	
	public MessageProcessor getMessageProcessor() {
		return messageProcessor;
	}

	protected DataIntegrateProcessor[] getDataIntegrateProcessors() {
		List<DataIntegrateProcessor> processors = new LinkedList<DataIntegrateProcessor>();
		processors.addAll(dataTransformers.values());
		processors.addAll(dataBroadcasters);
		processors.addAll(dataReceivers);
		return processors.toArray(new DataIntegrateProcessor[0]);
	}

	public boolean supportData(Class<? extends DataEdgeDescriptor> descriptorClazz) {
		Assert.notNull(descriptorClazz);
		DataIntegrateProcessor processors[] = getDataIntegrateProcessors();
		for (DataIntegrateProcessor processor : processors) {
			if (processor.supportData(descriptorClazz)) return true;
		}
		return false;
	}

	public boolean supportData(DataEdgeDescriptor dataDescriptor) {
		Assert.notNull(dataDescriptor);
		DataIntegrateProcessor processors[] = getDataIntegrateProcessors();
		for (DataIntegrateProcessor processor : processors) {
			if (processor.supportData(dataDescriptor)) return true;
		}
		return false;
	}

	public void process(DataIntegrateBatch batch) {
		Assert.notNull(batch);
		Assert.notNull(batch.getDataDescriptor());
		Assert.notNull(batch.getInputParameter());
		Assert.isInstanceOf(MessageEntity.class, batch.getInputParameter(), "批处理输入参数必须为MessageEntity类型。");
		DataEdgeDescriptor dataDescriptor = batch.getDataDescriptor();
		MessageEntity msgEntity = (MessageEntity) batch.getInputParameter();
		
		try {
			Assert.isTrue(doProcess(dataDescriptor, msgEntity), String.format("未发现集成数据(%s)对应的处理器。", dataDescriptor.getMessageCode()));
		} catch (Throwable e) {
			batch.setException(e);
		} finally {
			batch.setResult(null);
		}
	}
	
	protected boolean doProcess(DataEdgeDescriptor descriptor, MessageEntity msgEntity) {
		Class<? extends DataEdgeDescriptor> dataClazz = (Class<? extends DataEdgeDescriptor>) ClassUtils.getUserClass(descriptor);
		
		byte[] bytes = null;
		
		//先判断转换
		if (dataTransformers.containsKey(dataClazz)) {
			DataTransformer transformer = dataTransformers.get(dataClazz);
			bytes = transformer.transform(descriptor, msgEntity);
		} else {
			try {
				bytes = getMessageProcessor().toMessageBytes(msgEntity).toString().getBytes("utf-8");
			} catch (UnsupportedEncodingException e) {
				throw new IllegalArgumentException(e);
			}
		}

		boolean processed = false;
		//广播
		for (DataBroadcaster broadcaster: dataBroadcasters) {
			if (broadcaster.supportData(descriptor)) {
				broadcaster.broadcast(descriptor, bytes);
				processed = true;
			}
		}
		
		//接收
		for (DataReceiver dataMessageReceiver : dataReceivers) {
			if (dataMessageReceiver.supportData(descriptor)) {
				MessageEntity entity = null;;
				try {
					entity = getMessageProcessor().fromMessageBytes(bytes);
				} catch (MessageException e) {
					Log.warn("");
					continue;
				}
				dataMessageReceiver.receiveData(entity);
				processed = true;
			}
		}
		return processed;
	}

	public void process(DataEdgeDescriptor dataDescriptor, MessageEntity msgEntity) {
		Assert.notNull(msgEntity);
		Assert.isTrue(doProcess(dataDescriptor, msgEntity), String.format("未发现集成数据(%s)对应的处理器。", dataDescriptor.getMessageCode()));
	}

}
